Contentful API とPythonを使って記事を全件取得する。

Contentful API とPythonを使って記事を全件取得する。

Clock Icon2023.12.15

どうも、ベルリンオフィスの小西です。

ヘッドレスCMSのContentfulで、あるモデルの記事を全件取得する必要がありました。

ContentfulではAPIを通じて記事の一括取得(Entry Collection)が可能ですが、一度のリクエストで1,000件までしか取得できない制限があります。

ある程度の規模でサイトを作っていると1,000件以上の記事取得は頻繁にあるため、これを機に汎用的に使えるPythonを作ってみました。

前提

PythonからContentful APIへのリクエストには Python client library を使います。

セットアップは↓から。

https://www.contentful.com/developers/docs/python/tutorials/getting-started-with-contentful-and-python/

別で contentful-management という書き込み用ライブラリもありますが、今回は使いません。異なるメソッドを使っていたりするため注意が必要です。 Contenful APIの種別については以前↓にまとめています。

https://dev.classmethod.jp/articles/contentful-endpoint/

Pythonコード

早速ですがPythonコードです。

# -*- coding: utf-8 -*-
import contentful
import time

SPACE_ID = ""
CDA_TOKEN = ""
CONTENT_MODEL = ""

def get_client():
    # Contentfulクライアントを取得
    try:
        return contentful.Client(SPACE_ID, CDA_TOKEN)
    except Exception as e:
        print(f"Error while getting client: {e}")
        return None

def get_entries(client, content_type, limit, skip):
    max_retries = 3
    retries = 0

    # 指定された条件でエントリを取得、失敗時に3回までリトライ
    while retries < max_retries:
        try:

            return client.entries({
                'content_type': CONTENT_MODEL,
                #'select': 'sys.id,fields.title,fields.slug', #取得フィールドの選択
                'limit': limit,
                'skip': skip
            })
        except Exception as e:
            print(f"Error while getting entries: {e}")
            retries += 1
            if retries < max_retries:
                print(f"Retrying in 1 second (retry {retries} of {max_retries})...")
                time.sleep(2)
            else:
                print("Max retries reached. Exiting.")
                return []

def main():
    client = get_client()
    limit = 1000 # 一度に取得する件数 max:1000 複雑な条件ではタイムアウト1sにHITすることがある

    try:
        # 全エントリ数を取得
        total_entries = client.entries({
            'content_type': CONTENT_MODEL,
            'limit': 1
        }).total
        print("Total:", total_entries)

        # 必要な取得回数を計算
        skip_max = -(-total_entries // limit)

        num = 1
        for skip in range(skip_max):
            entries = get_entries(client, content_type, limit, skip * limit)

            for entry in entries:
                # 取得した記事に対する処理: 下記例ではslugを取り出している
                print(num, getattr(entry, 'slug'))

                num += 1
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

if __name__ == "__main__":
    main()

実際に記事を取得しているのはclient.entries()の部分です。

その際のオプションとして、取得したい記事モデルをcontent_typeとして指定しますが、それ以外にもAPIが受け入れるクエリパラメーターを指定できます。

例:

products_by_price = client.entries({
    'content_type': '<product_content_type_id>',
    'order': 'fields.price',  #priceフィールドの値で降順にソート
    'limit': 1000, #1,000件まで取得
    'fields.title[in]': 'example' #titleフィールドに'example'文字列が含む
})

1,000件以上の記事を取得する際にはskipオプションを使います。

例:

client.entries({
    'content_type': content_type,
    'skip': 1000
})

skipでは、記事を取得開始する開始位置を指定(オフセット)できます。

デフォルトは0で、例えば skip = 100 とした場合は、最初の100件の記事はスキップされ、101件目以降の記事をレスポンスに含むようになります(記事の並び順は order オプションによる)。

また記事のコレクションを取得した場合レスポンスには total が格納されています。これで記事の総数が把握できます。

そのため上記コードでは

  1. 最初に記事の総件数を取得
  2. 記事の総件数から必要な取得回数を計算
  3. 必要な取得回数まで skip を増やしていく

という流れで全件ループ取得しています。

注意: APIのタイムアウト

一度の記事取得上限は1,000件ですが、クエリ条件が複雑な場合は処理に1秒以上かかり、結果エラーとなることがあります。

例えば order オプションや、特定フィールドの値での絞り込みなどは負荷を上げやすく、その場合は limit オプションの値を下げることで回避できる可能性があります。

上記のコードでは稀に失敗することを想定してリトライ処理を入れています。

最後に

あくまでContentfulのREST APIの仕様に沿った実装のため、基本的な部分は他の言語やcurlなどでも流用できるかと思います。

クラスメソッドではContentfulの契約のご相談、構築支援をしています。ご興味のある方はぜひ弊社までお問い合わせください。

参考資料

Share this article

facebook logohatena logotwitter logo

© Classmethod, Inc. All rights reserved.